package filter.sql;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.Charset;

/**
 * filter包,过滤消息,消费时，选择指定的消息消费。
 * tag,根据发送消息时的tag过滤
 * sql,根据发送消息时设置的属性message.putUserProperty进行过滤，用SQL语法描述。
 *
 * @author itbac
 * @email 1218585258@qq.com
 * @date 2020/12/13
 */
public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建消息生产者 producer，并指定生产者组名 group1
        DefaultMQProducer producer = new DefaultMQProducer("group8");
        //2.指定Nameserver地址,集群用分号;隔开 。
        // （注意防火墙要开放指定端口,Nameserver端口9876 ,broker端口10911,broker要设置配置文件启动）
        producer.setNamesrvAddr("192.168.0.88:9876");
        //3.启动producer
        producer.start();

        //4.创建消息对象，指定主题Topic、Tag 和消息体
        /**
         *参数1：topic主题 （大分类）
         *参数2：tag （小分类）
         *参数3：body （消息内容）
         */
        //组装批量消息
        for (int i = 0; i < 10; i++) {
            Message message = new Message("filterSql", "tag1",
                    ("rocketmq 发送过滤消息SQL").getBytes(Charset.forName("UTF-8")));
            //给消息设置属性abc
            message.putUserProperty("abc", String.valueOf(i));
            //5.发送批量消息,拿到返回结果.
            SendResult result = producer.send(message);
            System.out.println(result);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}
